package  com.qyer.log.job.sum;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import com.qyer.util.log.entity.AppTrackLog;
import com.qyer.util.log.entity.TrackLog;
import com.qyer.util.log.service.LogService;
import com.qyer.util.log.service.impl.LogServiceImpl;
import com.qyer.util.log.util.LogFatory;
import com.qyer.util.log.util.MyDateUtil;
import com.qyer.util.log.util.PlatformUtil;
import com.qyer.util.log.util.ProductTypeUtil;
import com.qyer.util.log.util.SuppliersUtil;
import com.qyer.util.log.util.TrackSplitUtils;

public class ZTrackSummary {

	public static class M1 extends Mapper<LongWritable, Text, Text, Text> {
		private Text keyVal = new Text();
		private Text val = new Text();
		
		static HashMap<String, String> productMap= new HashMap<String, String>();
		static HashMap<String, String> sidMap= new HashMap<String, String>();
		static HashMap<String, String> isSelfMap= new HashMap<String, String>();
		static HashMap<String, String> groupMap= new HashMap<String, String>();
		static HashMap<String, String> tagMap= new HashMap<String, String>();
		static HashMap<String, String> sgidMap= new HashMap<String, String>();
		static List<String > applist = new ArrayList<String>();
		

		@Override
		protected void setup(Context context) throws IOException, InterruptedException {
			BufferedReader in = null;

			try {
				// 从当前作业中获取要缓存的文件
				Path[] paths = DistributedCache.getLocalCacheFiles(context.getConfiguration());
				
				System.out.println("paths===="+paths);
				String line = null;
				String[] strArr = null;
				for (Path path : paths) {
					System.out.println("path===="+path.toString());
					
					if (path.toString().contains("pt_type.txt")) {
						in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()),"utf-8")));
						while (null != (line = in.readLine())) {
							strArr = line.split("\t", -1);
							
							if(strArr.length<3){
								continue ;
							}
							
							productMap.put(strArr[0], strArr[1]+","+strArr[2]);
							sidMap.put(strArr[0], strArr[3]);
						}
						
					}else if (path.toString().contains("sid_type.txt")) {
						in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()),"utf-8")));
						while (null != (line = in.readLine())) {
							strArr = line.split("\t", -1);
							
							if(strArr.length<3){
								continue ;
							}
							
							isSelfMap.put(strArr[0], strArr[1]);
							groupMap.put(strArr[0], strArr[2]);
						}
						
					}else if (path.toString().contains("tag_type.txt")) {
						in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()),"utf-8")));
						while (null != (line = in.readLine())) {
							strArr = line.split("\t", -1);
							
							if(strArr.length<2){
								continue ;
							}
							
							tagMap.put(strArr[1], strArr[0]);
						}
						
					}else if (path.toString().contains("sgid_type.txt")) {
						in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()),"utf-8")));
						while (null != (line = in.readLine())) {
							strArr = line.split("\t", -1);
							
							if(strArr.length<2){
								continue ;
							}
							
							sgidMap.put(strArr[1], strArr[0]);
						}
						
					}else if (path.toString().contains("applist.txt")) {
						in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()),"utf-8")));
						while (null != (line = in.readLine())) {
							strArr = line.split("\t", -1);
							
							if(strArr.length<1){
								continue ;
							}
							applist.add(strArr[0]);
						}
						
					}
				}
				
				if (in != null) {
					in.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			} finally {
				try {
					if (in != null) {
						in.close();
					}
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}

		final static String FS="\t";
		
		static Pattern p2=Pattern.compile("\"id\";s:[0-9]+:\"([0-9]+)\";");
		
		//session 切分工具类
		TrackSplitUtils trackSplit = TrackSplitUtils.getInstance();
		//清洗日志工具类
		LogService lli = (LogService)LogFatory.getInstance(LogServiceImpl.class.getName());
		
		TrackLog tracklog = null ;
		
		AppTrackLog ll = null ; 

		public void map(LongWritable key, Text value, Context context) throws IOException,InterruptedException {
			InputSplit inputSplit = context.getInputSplit();
			String path = ((FileSplit) inputSplit).getPath().toString();
			String platform = "";
			
			//app
			if (path.indexOf("apptrack")>-1){
				platform=PlatformUtil.APP;
				
				ll =lli.getAppTrackFormat(value.toString());
				
				if(ll==null){
					return ;
				}
				
				String client_id = ll.getClientId();
				if(client_id.equals("qyer_discount_androi")){
					client_id="qyer_discount_android";
				}
				String api_name=ll.getApiName();
				String info =ll.getInfo();
				String track_deviceid=ll.getDeviceid();
				String type="";
				
				if(!applist.contains(client_id)){
					return; 
				}
				
				if(!"/lastminute/get_detail".equals(api_name)&&!"/lastminute/app_get_productsinfo".equals(api_name)){
					return ; 
				}
				
				if("/lastminute/get_detail".equals(api_name)){
					type="1"; 
				}else if(api_name.indexOf("/lastminute/app_get_productsinfo")>-1){
					type="2"; 
				}
				
				
				List <String>pcdo=PlatformUtil.getPlatform(platform ,  client_id ,null) ;
				String lid ="";
				
				Matcher m=p2.matcher(info);  
				while(m.find()){  
					lid=m.group(1);  
				} 
				
				if("".equals(lid.trim())){
					return ;
				}
				
				String producttype=productMap.get(lid);
				
				if(producttype==null){
					return ; 
				}
				
				List<String> pl = ProductTypeUtil.getDimList2(lid , producttype);
				
				String sid =sidMap.get(lid);
				
				String isSelf=isSelfMap.get(sid);
				
				String tagStr =tagMap.get(lid);
				
				String group =groupMap.get(sid);
				
				String sgid =sgidMap.get(lid);
				
				List<String> sl = SuppliersUtil.getDimList(isSelf , group , sgid , sid);
				
				
				List<String> tags = new ArrayList <String>();
				tags.add("A"); 
				if(tagStr!=null){
					for(String ts : tagStr.split(",")){
						tags.add(ts); 
					}
				}
				Text val = new Text();
				Text keyVal =new Text();
				
				for(String t : new String []{"A"  , type}){
					for (String p : pcdo){
			    		for(String pts : pl) {
			    			for(String ta : tags) {
			    				for(String sss :sl){
    								String keyStr = t+FS+p+FS+pts+FS+sss+FS+ta; 
					    			String valStr = track_deviceid;
					    			keyVal.set(keyStr);
					    			val.set(valStr);
					    			context.write(keyVal,val);    
			    				}
			    			}	
						}
				    }
				}
			// web m apph5	
			}else{
				tracklog= lli.getTrackFormat(value.toString());
				
				if(tracklog==null){
					return ;
				}
				
				platform=tracklog.getPlatform();
				
				String type=tracklog.getType();
				
				if(!"1".equals(type)&&!"2".equals(type)){
					return ; 
				}
				
				
				String category=tracklog.getCategory();
				if ("2".equals(type)){
					if (!"m-lm-paydetail-booking".equals(category)&&!"lm-detail-booking-main".equals(category)&&!"lm-detail-booking-bar".equals(category)&&!"normal-booking".equals(category)){
						return ; 
					}
				}
				
				String ugctype=tracklog.getUgc_type();
				if(ugctype.indexOf("detail")<0){
					return ; 
				}else{
					ugctype="detail";
				}
				
				
				String url=tracklog.getCurUrl();
				String os_browser =tracklog.getUserAgent();
				String lid=tracklog.getUgc_content();
				List<String> pcdo= null ; 
				
				platform = tracklog.getPlatform();
				
				if(PlatformUtil.APPH5.equals(platform)){
					String clientId = url.substring(url.indexOf("?") +1);
					Map<String , String> urlMaps = new HashMap<String,String>();
					for(String kv : clientId.split("&")){
						if(kv.indexOf("=")>-1){
							String[] kvs = kv.split("=");
							if(kvs.length>1){
								urlMaps.put(kvs[0] , kvs[1]);
							}
						}
					}
					
					clientId=urlMaps.get("client_id");
					if(clientId==null){
						return ;
					}
					
					if(!clientId.equals("qyer_planner_ios")&&!clientId.equals("qyer_planner_android")){
						return ; 
					}
					pcdo=PlatformUtil.getPlatform(platform , clientId ,os_browser) ;
				}else if (PlatformUtil.WEB.equals(platform)||PlatformUtil.MOBILE.equals(platform)){
					pcdo=PlatformUtil.getPlatform(platform ,  null ,os_browser ) ;
				}
				
				String unique_id=tracklog.getUnique_id();
				
				if(StringUtils.isEmpty(unique_id)){
					return; 
				}
				
				String producttype=productMap.get(lid);
				
				if(producttype==null){
					return ; 
				}
				
				List<String> pl = ProductTypeUtil.getDimList2(lid , producttype);
				
				String sid =sidMap.get(lid);
				
				String isSelf=isSelfMap.get(sid);
				
				String tagStr =tagMap.get(lid);
				
				String group =groupMap.get(sid);
				
				String sgid =sgidMap.get(lid);
				
				List<String> sl = SuppliersUtil.getDimList(isSelf , group , sgid , sid);
				
				
				List<String> tags = new ArrayList <String>();
				tags.add("A"); 
				if(tagStr!=null){
					for(String ts : tagStr.split(",")){
						tags.add(ts); 
					}
				}
				
				for(String t : new String []{"A"  , type}){
					for (String p : pcdo){
			    		for(String pts : pl) {
			    			for(String ta : tags) {
			    				for(String sss :sl){
    								String keyStr = t+FS+p+FS+pts+FS+sss+FS+ta; 
					    			String valStr = unique_id;
					    			keyVal.set(keyStr);
					    			val.set(valStr);
					    			context.write(keyVal, val);
			    				}
			    			}	
						}
				    }
				}
			}
			
		}
	}

	
	public static class R1 extends Reducer<Text, Text, Text, Text> {
		 private Text val = new Text();
		 private Text keyVal =new Text();
		  
		 final static String FS="\t";
		 final static String NFS="\001";
		  
		 static String DATE ;
		  
		 static String len;
		
		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException,
				InterruptedException {

			if (DATE == null) {
				DATE = context.getConfiguration().get("DATE");
			} 
 			
 			if (len == null) {
 				len = context.getConfiguration().get("len");
			}
 			 
  			if(key.toString().contains("null")){
  				return ; 
  			} 
        	  
  			Set<String> cookieIdSet = new HashSet<String>();
  			long uv = 0l;
  			long pv = 0l;

  			for (Text val : values) {
				pv++; 
				cookieIdSet.add(val.toString()); 
			}

  			uv=cookieIdSet.size();
			
			String keyv=key.toString().replace(FS, NFS);
			keyVal.set(len+NFS+DATE+NFS+keyv+NFS+pv+NFS+uv);
			context.write(keyVal, val);
		}
	}

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException,
			URISyntaxException, ParseException {

		Configuration conf = new Configuration();
//		conf.set("mapred.job.queue.name", "regular"); // default,regular,realtime
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		int exitCode = 127;
		
		if (otherArgs.length < 1) {
			System.out.println("************************************************************");
			System.out.println("************************************************************");
			System.out.println("Usage: please input 1 params, for example: file.jar args[0]");
			System.out.println("args[0] is dateList: 20130101,20130106 or 20130106");
			System.out.println("************************************************************");
			System.out.println("************************************************************");
			System.exit(exitCode);
		}
		
		String startDate = otherArgs[0].split(",")[0];
		String endDate = otherArgs[0].split(",").length == 2 ? otherArgs[0].split(",")[1] : startDate;
		
		String len =otherArgs[1]; 

		String baseWeb = "/flume/clean/track/";
		String baseApp ="/flume/clean/apptrack/";
		String baseOutPath = "/flume/summary/z_track_2/";
		
		String ptPath="/flume/dict/product_type";
		String sidPath="/flume/dict/sid_type";
		String tagPath="/flume/dict/tag_type";
		String sgidPath="/flume/dict/sgid_type";
		String applistPath=otherArgs[2];
		
		String runDate= startDate;
		
		if(len.equals("1")){
			conf.set("mapreduce.reduce.memory.mb" ,"3048");
			conf.set("len", "1");
			endDate=startDate;
		}else if(len.equals("7")){
			conf.set("mapreduce.reduce.memory.mb" ,"6048");
			endDate=MyDateUtil.getNDaysAgo(startDate , -6);
			conf.set("len", "7");
		}else if(len.equals("30") || len.equals("28") ||len.equals("31")){
			conf.set("mapreduce.reduce.memory.mb" ,"6048");
			endDate=MyDateUtil.getNDaysAgo(startDate , -Integer.parseInt(len));
			conf.set("len", "30");
		}
		

		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
		Long startDateLong =sdf.parse(runDate).getTime();
		startDateLong=startDateLong/1000;
		conf.set("DATE", startDateLong.toString());
		
		Job job = new Job(conf, "ZTrackSummary2");
		job.setInputFormatClass(TextInputFormat.class);
		job.setJarByClass(ZTrackSummary.class);
		job.setMapperClass(M1.class);
		job.setReducerClass(R1.class);
		job.setNumReduceTasks(3);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		
        // source type 字典
		FileSystem hdfs1 = FileSystem.get(conf);
		
		Integer rd =Integer.parseInt(runDate); 
		String ptDate=runDate;
		if (rd < 20151217){
			ptDate="20151217";
		}
		
		FileStatus[] fileList = hdfs1.listStatus(new Path(ptPath+"/"+ptDate));
		for (int i = 0; i < fileList.length; i++) {
			String fileName = fileList[i].getPath().getName();
			if (fileName.contains("pt_type.txt")) {
				System.out.println("----------------->>> " + ptPath+"/"+ptDate + "/" + fileName);
				String inPath2Link = new Path(ptPath+"/"+ptDate + "/" + fileName).toUri().toString() + "#"+ "pt_type.txt" + i;
				DistributedCache.addCacheFile(new URI(inPath2Link), job.getConfiguration());
			}
		}
		
		String sidDate=runDate;
		if (rd < 20151228){
			sidDate="20151228";
		}
		
		fileList = hdfs1.listStatus(new Path(sidPath+"/"+sidDate));
		for (int i = 0; i < fileList.length; i++) {
			String fileName = fileList[i].getPath().getName();
			if (fileName.contains("sid_type.txt")) {
				System.out.println("----------------->>> " + sidPath+"/"+sidDate + "/" + fileName);
				String inPath2Link = new Path(sidPath+"/"+sidDate + "/" + fileName).toUri().toString() + "#"+ "sid_type.txt" + i;
				DistributedCache.addCacheFile(new URI(inPath2Link), job.getConfiguration());
			}
		}
		
		
		String tagDate=runDate;
		if (rd < 20151228){
			tagDate="20151228";
		}
		
		fileList = hdfs1.listStatus(new Path(tagPath+"/"+tagDate));
		for (int i = 0; i < fileList.length; i++) {
			String fileName = fileList[i].getPath().getName();
			if (fileName.contains("tag_type.txt")) {
				System.out.println("----------------->>> " + tagPath+"/"+tagDate + "/" + fileName);
				String inPath2Link = new Path(tagPath+"/"+tagDate + "/" + fileName).toUri().toString() + "#"+ "tag_type.txt" + i;
				DistributedCache.addCacheFile(new URI(inPath2Link), job.getConfiguration());
			}
		}
		
		
		String sgidDate=runDate;
		if (rd < 20160110){
			sgidDate="20160110";
		}
		
		fileList = hdfs1.listStatus(new Path(sgidPath+"/"+sgidDate));
		for (int i = 0; i < fileList.length; i++) {
			String fileName = fileList[i].getPath().getName();
			if (fileName.contains("sgid_type.txt")) {
				System.out.println("----------------->>> " + sgidPath+"/"+sgidDate + "/" + fileName);
				String inPath2Link = new Path(sgidPath+"/"+sgidDate + "/" + fileName).toUri().toString() + "#"+ "sgid_type.txt" + i;
				DistributedCache.addCacheFile(new URI(inPath2Link), job.getConfiguration());
			}
		}
		
		fileList = hdfs1.listStatus(new Path(applistPath));
		for (int i = 0; i < fileList.length; i++) {
			String fileName = fileList[i].getPath().getName();
			if (fileName.contains("applist.txt")) {
				System.out.println("----------------->>> " + applistPath+ "/" + fileName);
				String inPath2Link = new Path(applistPath+ "/" + fileName).toUri().toString() + "#"+ "applist.txt" + i;
				DistributedCache.addCacheFile(new URI(inPath2Link), job.getConfiguration());
			}
		}
		
		for (String rDate : MyDateUtil.getDateList(startDate, endDate, 0)) {
			System.out.println("runDate -------------->" +rDate);
			FileInputFormat.addInputPath(job, new Path(baseWeb + rDate));
			FileInputFormat.addInputPath(job, new Path(baseApp + rDate));
		}
		
// 			conf.set("PLATFORM", p);
        
        String outPath = baseOutPath +"/"+ runDate;
        FileSystem.get(conf).delete(new Path(outPath), true);
        FileOutputFormat.setOutputPath(job, new Path(outPath));
        
        //启用压缩
        FileOutputFormat.setCompressOutput(job, true);
        //lzo格式
        FileOutputFormat.setOutputCompressorClass(job, com.hadoop.compression.lzo.LzopCodec.class); 
        
        exitCode = job.waitForCompletion(true) ? 0 : 1;
        System.out.println("--------------------------------------------END1"+runDate);
	        

	}

}
